package com.noahsloan.nutils.streams.clocked;

/**
 * {@link RateCalculator} that uses a sliding window to calculate the data rate.
 * e.g. for an interval of 1 second, all bytes read in the last 1 second are
 * divided by the 1000ms in 1 second to get the rate.
 * 
 * @author noah
 * 
 */
public class SlidingWindowCalc implements Clocked, RateCalculator {

    private int heartbeatCount;

    private final int cleanTick;

    /**
     * Simple linked list of readings. This one always points to the most recent
     * reading.
     */
    private Reading readings;

    private long lastRead;

    private long end;

    private TimeSource source;

    public SlidingWindowCalc(long interval, TimeSource source) {
        super();
        lastRead = System.currentTimeMillis();
        this.cleanTick = (int) (Math.ceil(interval / source.getTick()) + 1);
        this.source = source;
        source.register(this);
    }

    public void heartbeat() {
        // record that 0 bytes have been read since the last read
        storeMeasurement(0);

        // every interval, cut off the extra measurements
        ++heartbeatCount;
        if ((heartbeatCount %= cleanTick) == 0) {
            long minTime = end == 0 ? source.currentTime() : end;
            Reading r = readings;
            while (r.next != null && r.next.end < minTime) {
                r = r.next;
            }
            // cut off the measurements before the interval started
            r.next = null;
        }
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.nutils.streams.clocked.RateCalculator#storeReading(long)
     */
    public synchronized long storeMeasurement(final long count) {
        long rCount = Math.max(0, count);
        long time = source.currentTime();
        Reading r = new Reading(lastRead, time, rCount);
        r.next = readings;
        readings = r;
        lastRead = time;
        return count;
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.nutils.streams.clocked.RateCalculator#close()
     */
    public void close() {
        end = source.currentTime();
        source.unregister(this);
    }

    /*
     * (non-Javadoc)
     * 
     * @see org.nutils.streams.clocked.RateCalculator#getRate()
     */
    public synchronized double getRate() {
        double byteCount = 0;
        double totalTime = 0;

        Reading r = readings;
        while (r != null) {
            byteCount += r.count;
            totalTime += r.end - r.start;
            r = r.next;
        }

        return byteCount / totalTime;
    }

    private static class Reading {
        /**
         * Start time.
         */
        public final long start;

        /**
         * End time.
         */
        public final long end;

        /**
         * Number of bytes that have been read.
         */
        public long count;

        /**
         * The next reading.
         */
        public Reading next;

        public Reading(final long start, final long end, final long count) {
            super();
            this.start = start;
            this.end = end;
            this.count = count;
        }
    }

    @Override
    protected void finalize() throws Throwable {
        close();
        super.finalize();
    }

}
